import {
  Injectable,
  OnModuleInit,
  OnModuleDestroy,
  Logger,
} from '@nestjs/common';
import * as amqp from 'amqplib';

@Injectable()
export class RabbitmqService implements OnModuleInit, OnModuleDestroy {
  private client: amqp.Connection;
  private channel: amqp.Channel;
  private readonly logger = new Logger(RabbitmqService.name);

  // 模块初始化时，建立 RabbitMQ 连接
  async onModuleInit() {
    try {
      // 建立连接
      this.client = await amqp.connect(`amqp://localhost:5672`);

      // 创建一个通信通道
      this.channel = await this.client.createChannel();

      this.logger.log('RabbitMQ 连接已建立');
    } catch (error) {
      this.logger.error('RabbitMQ 连接错误:', error);
      // 根据需要，可以在此处重试连接或抛出异常
    }
  }

  // 发送消息的方法
  async send(queue: string, message: any): Promise<void> {
    try {
      if (!this.channel) {
        throw new Error('通道尚未初始化');
      }

      // 在发送消息之前，确保队列存在
      await this.channel.assertQueue(queue, { durable: true });

      // 发送消息到队列
      this.channel.sendToQueue(queue, Buffer.from(JSON.stringify(message)), {
        persistent: true, // 确保消息持久化，以防服务器故障
      });

      this.logger.log(`消息已发送到队列 ${queue}: ${JSON.stringify(message)}`);
    } catch (error) {
      this.logger.error('发送消息到 RabbitMQ 时出错:', error);
    }
  }

  // 消费消息的方法
  async consume(queue: string, onMessage: (msg: any) => void): Promise<void> {
    try {
      if (!this.channel) {
        throw new Error('通道尚未初始化');
      }

      // 确保队列存在
      await this.channel.assertQueue(queue, { durable: true });

      // 设置预取计数为1，确保每次只处理一个消息
      await this.channel.prefetch(1);

      // 开始消费消息
      await this.channel.consume(
        queue,
        async (msg) => {
          if (msg !== null) {
            try {
              const messageContent = JSON.parse(msg.content.toString());
              this.logger.log(
                `收到来自队列 ${queue} 的消息: ${JSON.stringify(messageContent)}`,
              );

              // 调用传入的回调函数处理消息
              await onMessage(messageContent);

              // 手动确认消息已被处理
              this.channel.ack(msg);
            } catch (processingError) {
              this.logger.error('处理消息时出错:', processingError);
              // 根据需要，可以选择不确认消息，以便重新投递
              this.channel.nack(msg, false, false); // 第二个参数表示是否批量确认，第三个参数表示是否重新入队
            }
          }
        },
        {
          noAck: false, // 确保手动确认消息
        },
      );

      this.logger.log(`开始消费队列 ${queue}`);
    } catch (error) {
      this.logger.error('消费消息时出错:', error);
    }
  }

  // 模块销毁时，关闭连接和通道
  async onModuleDestroy() {
    try {
      if (this.channel) {
        await this.channel.close();
      }
      if (this.client) {
        await this.client.close();
      }
      this.logger.log('RabbitMQ 连接已关闭');
    } catch (error) {
      this.logger.error('关闭 RabbitMQ 连接时出错:', error);
    }
  }
}
